package org.xq.softcup.mq.util;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@Slf4j
public class ZkClient {
    private String zkaddress;
    private String zkpath;
    private String zkdigest;
    private Watcher watcher;
    private ZooKeeper zooKeeper;

    public ZkClient(String zkaddress, String zkpath, String zkdigest, Watcher watcher)
    {
        this.zkaddress = zkaddress;
        this.zkpath = zkpath;
        this.zkdigest = zkdigest;
        this.watcher = watcher;
        if (this.watcher == null) {
            this.watcher = new Watcher()
            {
                public void process(WatchedEvent watchedEvent)
                {
                    ZkClient.log.info("监控:{}", watchedEvent);
                    if (watchedEvent.getState() == Watcher.Event.KeeperState.Expired)
                    {
                        ZkClient.this.destroy();
                        ZkClient.this.getClient();
                    }
                }
            };
        }
    }

    private ReentrantLock INSTANCE_INIT_LOCK = new ReentrantLock(true);

    public ZooKeeper getClient()
    {
        if (this.zooKeeper == null) {
            try
            {
                if (this.INSTANCE_INIT_LOCK.tryLock(2L, TimeUnit.SECONDS))
                {
                    ZooKeeper newZk = null;
                    try
                    {
                        if (this.zooKeeper == null)
                        {
                            newZk = new ZooKeeper(this.zkaddress, 10000, this.watcher);
                            if ((this.zkdigest != null) && (this.zkdigest.trim().length() > 0)) {
                                newZk.addAuthInfo("digest", this.zkdigest.getBytes());
                            }
                            newZk.exists(this.zkpath, false);


                            this.zooKeeper = newZk;
                            log.info("ZkClient 初始化成功.");
                        }
                    }
                    catch (Exception e)
                    {
                        if (newZk != null) {
                            newZk.close();
                        }
                        log.error(e.getMessage(), e);
                    }
                    finally
                    {
                        this.INSTANCE_INIT_LOCK.unlock();
                    }
                }
            }
            catch (Exception e)
            {
                log.error(e.getMessage(), e);
            }
        }
        if (this.zooKeeper == null) {
            throw new XQRpcException("ZkClient.zooKeeper 为空.");
        }
        return this.zooKeeper;
    }

    public void destroy()
    {
        if (this.zooKeeper != null) {
            try
            {
                this.zooKeeper.close();
                this.zooKeeper = null;
            }
            catch (Exception e)
            {
                log.error(e.getMessage(), e);
            }
        }
    }

    private Stat createPathWithParent(String path, boolean watch)
    {
        if ((path == null) || (path.trim().length() == 0)) {
            return null;
        }
        try
        {
            Stat stat = getClient().exists(path, watch);
            if (stat == null)
            {
                if (path.lastIndexOf("/") > 0)
                {
                    String parentPath = path.substring(0, path.lastIndexOf("/"));
                    Stat parentStat = getClient().exists(parentPath, watch);
                    if (parentStat == null) {
                        createPathWithParent(parentPath, false);
                    }
                }
                getClient().create(path, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
            return getClient().exists(path, true);
        }
        catch (Exception e)
        {
            throw new XQRpcException(e);
        }
    }

    public void deletePath(String path, boolean watch)
    {
        try
        {
            Stat stat = getClient().exists(path, watch);
            if (stat != null) {
                getClient().delete(path, stat.getVersion());
            } else {
                log.info("zookeeper 节点路径无法找到 :{}", path);
            }
        }
        catch (Exception e)
        {
            throw new XQRpcException(e);
        }
    }

    public Stat setPathData(String path, String data, boolean watch)
    {
        try
        {
            Stat stat = getClient().exists(path, watch);
            if (stat == null)
            {
                createPathWithParent(path, watch);
                stat = getClient().exists(path, watch);
            }
            return getClient().setData(path, data.getBytes("UTF-8"), stat.getVersion());
        }
        catch (Exception e)
        {
            throw new XQRpcException(e);
        }
    }

    public String getPathData(String path, boolean watch)
    {
        try
        {
            String znodeValue = null;
            Stat stat = getClient().exists(path, watch);
            if (stat != null)
            {
                byte[] resultData = getClient().getData(path, watch, null);
                if (resultData != null) {
                    znodeValue = new String(resultData, "UTF-8");
                }
            }
            else
            {
                log.info("[{}]路径找不到.", path);
            }
            return znodeValue;
        }
        catch (Exception e)
        {
            throw new XQRpcException(e);
        }
    }

    public void setChildPathData(String path, String childNode, String childNodeData)
    {
        try
        {
            createPathWithParent(path, false);



            String childNodePath = path.concat("/").concat(childNode);

            Stat stat = getClient().exists(childNodePath, false);
            if (stat != null) {
                if (stat.getEphemeralOwner() == 0L) {
                    getClient().delete(childNodePath, stat.getVersion());
                } else {
                    return;
                }
            }
            getClient().create(childNodePath, childNodeData.getBytes("UTF-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        }
        catch (Exception e)
        {
            throw new XQRpcException(e);
        }
    }

    public void deleteChildPath(String path, String childNode)
    {
        try
        {
            String childNodePath = path.concat("/").concat(childNode);
            deletePath(childNodePath, false);
        }
        catch (Exception e)
        {
            throw new XQRpcException(e);
        }
    }

    public Map<String, String> getChildPathData(String path)
    {
        Map<String, String> allData = new HashMap();
        try
        {
            Stat stat = getClient().exists(path, true);
            if (stat == null) {
                return allData;
            }
            List<String> childNodes = getClient().getChildren(path, true);
            if ((childNodes != null) && (childNodes.size() > 0)) {
                for (String childNode : childNodes)
                {
                    String childNodePath = path.concat("/").concat(childNode);
                    String childNodeValue = getPathData(childNodePath, false);

                    allData.put(childNode, childNodeValue);
                }
            }
            return allData;
        }
        catch (Exception e)
        {
            throw new XQRpcException(e);
        }
    }
}

